package blue.mqtt4j.internal.client.net;

import blue.mqtt4j.client.MqttOptions;
import blue.mqtt4j.core.handler.SenderListener;
import blue.mqtt4j.core.util.ClientIdGen;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttConnAckMessage;
import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttMessageBuilders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;

/**
 * @author Jin Zheng
 * @since 2020-10-02
 */
public class MqttConnectHandler extends SimpleChannelInboundHandler<MqttConnAckMessage>
{
	private static Logger logger = LoggerFactory.getLogger(MqttConnectHandler.class);

	private final NettyClient client;
	private final MqttOptions options;

	public MqttConnectHandler(NettyClient client, MqttOptions options)
	{
		this.client = client;
		this.options = options;
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, MqttConnAckMessage message) throws Exception
	{
		MqttConnAckVariableHeader header = message.variableHeader();
		if (header.connectReturnCode() != MqttConnectReturnCode.CONNECTION_ACCEPTED)
		{
			logger.error("MQTT connect failure: {}", header.connectReturnCode());
			return;
		}
		Channel channel = ctx.channel();
		logger.debug("MQTT connect successful");
		client.onConnected(channel, header.isSessionPresent());
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception
	{
		MqttMessageBuilders.ConnectBuilder builder = MqttMessageBuilders.connect();
		builder.cleanSession(true)
				.keepAlive(options.keepAliveSecs())
				.clientId(options.clientId() == null || options.clientId().isEmpty() ?
							ClientIdGen.getInstance().clientId() : options.clientId());

		String username = options.username();
		if (username == null || username.isEmpty())
		{
			builder.hasUser(false);
		}
		else
		{
			builder.hasUser(true)
					.username(username);
		}
		String password = options.password();
		if (password == null || password.isEmpty())
		{
			builder.hasPassword(false);
		}
		else
		{
			builder.hasPassword(true)
					.password(password.getBytes(StandardCharsets.UTF_8));
		}

		if (options.willTopic() == null || options.willTopic().isEmpty())
		{
			builder.willFlag(false);
		}
		else
		{
			builder.willFlag(true)
					.willTopic(options.willTopic())
					.willQoS(options.willQoS().toNettyMqttQoS())
					.willMessage(options.willMessage())
					.willRetain(options.willRetain());
		}

		MqttConnectMessage message = builder.build();
		ctx.writeAndFlush(message).addListener(new SenderListener(message));
		logger.debug("MQTT connecting...");
		super.channelActive(ctx);
	}
}
